package com.smartboot.plus.features.connect;


import com.smartboot.plus.holder.BrokerContextHolder;
import com.smartboot.plus.ignite.cache.MqttClientCache;
import com.smartboot.plus.ignite.cache.MqttNodeCache;
import com.smartboot.plus.service.MqttClientService;
import com.smartboot.plus.utils.Func;
import tech.smartboot.feat.cloud.annotation.Autowired;
import tech.smartboot.feat.cloud.annotation.Bean;
import tech.smartboot.feat.cloud.annotation.PostConstruct;
import tech.smartboot.feat.core.common.logging.Logger;
import tech.smartboot.feat.core.common.logging.LoggerFactory;
import tech.smartboot.mqtt.common.message.MqttConnectMessage;
import tech.smartboot.mqtt.plugin.spec.BrokerContext;
import tech.smartboot.mqtt.plugin.spec.bus.EventObject;
import tech.smartboot.mqtt.plugin.spec.bus.EventType;

import java.sql.Timestamp;


@Bean
public class MqttConnectFeatures {
    private static final Logger LOGGER = LoggerFactory.getLogger(MqttConnectFeatures.class);

    @Autowired
    private MqttClientService mqttClientService;

    @Autowired
    private BrokerContext brokerContext;


    @PostConstruct
    public void init() {

        //客户端连接
        brokerContext.getEventBus().subscribe(EventType.CONNECT, (eventType, object) -> {
            try {
                String clientId = object.getSession().getClientId();
                if (Func.isEmpty(clientId)) {
                    return;
                }
                if (clientId.toUpperCase().startsWith("CLUSTER_HYB_")) {
                    //跳过当前acl检查
                    return;
                }

                MqttConnectMessage message = object.getObject();
                if (Func.isEmpty(message)) {
                    return;
                }
                String userName = object.getObject().getPayload().userName();
                if (Func.isEmpty(userName)) {
                    return;
                }

                String ipAddress = getIpAddress(object);
                MqttClientCache mqttClientEntity = new MqttClientCache();

                //客户端id
                mqttClientEntity.setClientId(clientId);

                mqttClientEntity.setVersion(message.getVersion().name());
                mqttClientEntity.setKeepalive(message.getVariableHeader().keepAliveTimeSeconds());
                //客户端账号
                mqttClientEntity.setUsername(message.getPayload().userName());
                //节点ID
                mqttClientEntity.setNodeId(brokerContext.Options().getNodeId());
                //String hostIp = INetUtil.getHostIp();
                //IP地址
                mqttClientEntity.setIpAddress(ipAddress);
                //连接时间
                mqttClientEntity.setConnectedAt(new Timestamp(System.currentTimeMillis()));
                //mqttClientService.removeById(clientId)
                mqttClientService.getIgniteCache().put(clientId, mqttClientEntity);

            } catch (Exception e) {
                LOGGER.error("Failed to process CONNECT event for client: {}", e);
            }
        });


        //客户端断开删除数据
        brokerContext.getEventBus().subscribe(EventType.DISCONNECT, (eventType, object) -> {
            try {
                String clientId = object.getClientId();
                if (clientId != null) {
                    mqttClientService.removeById(clientId);
                }
            } catch (Exception e) {
                LOGGER.error("Failed to process DISCONNECT event : {}", e);
            }
        });
    }

    // 获取客户端 IP 地址
    private String getIpAddress(EventObject<MqttConnectMessage> object) {
        String ipAddress = "127.0.0.1";
        try {
            if (object.getSession() != null && object.getSession().getRemoteAddress() != null) {
                ipAddress = object.getSession().getRemoteAddress().getHostString();
            }
        } catch (Exception e) {

        }
        return ipAddress;
    }
}
